# !/usr/bin/python3
# -*- coding:utf-8 -*-
"""
@author: JHC000abc@gmail.com
@file: cron_monitor.py
@time: 2024/11/5 11:31 
@desc: 

"""
from tasks import app_celery
from db.db_utils import SqlalchemyDB
from celery.result import AsyncResult
from db.model import TaskInfo
from base import state_map


class Monitor(object):
    """

    """

    def __init__(self):
        self.sdb = SqlalchemyDB()

    def get_work_by_status(self):
        """

        :return:
        """
        sql = "select `task_id` from `task_info` where `status` not in ('2','3');"
        results = self.sdb.select(sql=sql)
        for task_ids in results:
            task_id = task_ids[0]
            yield task_id

    def check_result(self, task_id):
        """

        :param task_id:
        :return:
        """
        res = AsyncResult(task_id, app=app_celery)
        status = state_map.get(res.state)
        if not res.result:
            return
        redis_result = str(res.result).replace("'", '"') if res.failed() else res.result.get("ScriptResults")
        traceback = "" if not res.traceback else res.traceback
        if traceback != "":
            redis_result = ""
        select_results = self.sdb.select(TaskInfo, conditions={"task_id": task_id})
        if select_results:
            conditions = []
            for select_result in select_results:
                condition = {
                    "id": select_result.id,
                    "task_id": select_result.task_id,
                    "status": status,
                    "err": traceback,
                    "result": redis_result
                }
                conditions.append(condition)
            self.sdb.update(TaskInfo, conditions)


    def process(self):
        """

        :return:
        """
        for task_id in self.get_work_by_status():
            print("task_id",task_id)
            self.check_result(task_id)


if __name__ == '__main__':
    m = Monitor()
    m.process()
